package com.lgCore.config;
/**
 * Created by by on 2020/11/7 13:47
 */

import com.alibaba.fastjson.JSON;
import com.lgCore.factory.RpcServiceZkFactory;
import com.lgCore.model.BaseConnectionInfo;
import com.lgCore.provide.service.impl.CommonServiceServerImpl;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.state.ConnectionState;
import org.apache.curator.framework.state.ConnectionStateListener;
import org.apache.curator.retry.BoundedExponentialBackoffRetry;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.curator.utils.CloseableUtils;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.data.Stat;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;


/**
 * curator 配置中心
 *
 * @program: sxreader_rpc
 * @author: ligang
 * @create: 2020-11-07 13:47
 */
public class CuratorConfigCenter {
//    private static final Log LOGGER = LogFactory.getLog(CommonServiceServerImpl.class);

    private static final Logger LOGGER = LoggerFactory.getLogger(CommonServiceServerImpl.class);

    /**
     * zk连接地址
     */
    private String connectionString;

    /**
     * 根目录
     */
    private String zkRootPath = BaseConnectionInfo.getZkRootPath();

    /**
     * 会话超时时间（毫秒）
     */
    private int sessionTimeoutMs = BaseConnectionInfo.getSessionTimeoutMs();
    /**
     * 连接超时时间（毫秒）
     */
    private int connectionTimeoutMs = BaseConnectionInfo.getConnectionTimeoutMs();
    /**
     * 重试等待时间（毫秒）
     */
    private int baseSleepTimeMs = BaseConnectionInfo.getBaseSleepTimeMs();


    private int maxSleepTimeMs = BaseConnectionInfo.getMaxSleepTimeMs();

    /**
     * 重试次数
     */
    private int maxRetries = BaseConnectionInfo.getMaxRetries();

    public CuratorConfigCenter(int sessionTimeoutMs, int connectionTimeoutMs, int baseSleepTimeMs, int maxRetries, String connectionString, String zkRootPath) {
        this.sessionTimeoutMs = sessionTimeoutMs;
        this.connectionTimeoutMs = connectionTimeoutMs;
        this.baseSleepTimeMs = baseSleepTimeMs;
        this.maxRetries = maxRetries;
        this.connectionString = connectionString;
        this.zkRootPath = zkRootPath;
        initCuratorClient();
    }

    public CuratorConfigCenter(String connectionString, String zkRootPath) {
        this.connectionString = connectionString;
        this.zkRootPath = zkRootPath;
        initCuratorClient();
    }

    public CuratorConfigCenter(String connectionString) {
        this.connectionString = connectionString;
        initCuratorClient();
    }


    private static final Logger log = LoggerFactory.getLogger(CuratorConfigCenter.class);
    private static String CONFIGURATION_ROOT_PATH = "";
    private CuratorFramework client;


    private void initCuratorClient() {
        try {
            CONFIGURATION_ROOT_PATH = concatKey(zkRootPath);
//            RetryPolicy retryPolicy = new ExponentialBackoffRetry(baseSleepTimeMs, maxRetries);
            RetryPolicy retryPolicy = new BoundedExponentialBackoffRetry(baseSleepTimeMs,maxSleepTimeMs, maxRetries);
            this.client = CuratorFrameworkFactory.builder()
                    .connectString(connectionString)
                    .sessionTimeoutMs(sessionTimeoutMs)
                    .connectionTimeoutMs(connectionTimeoutMs)
                    .retryPolicy(retryPolicy)
                    .build();

            //断开重连
            client.getConnectionStateListenable().addListener(getListener());
            client.start();
            Stat stat = client.checkExists().forPath(CONFIGURATION_ROOT_PATH);
            if (null == stat) {
                client.create().withMode(CreateMode.PERSISTENT).forPath(CONFIGURATION_ROOT_PATH);
            }
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }

    private String concatKey(String key) {
        return CONFIGURATION_ROOT_PATH.concat("/").concat(key);
    }

    public CuratorFramework getClient() {
        return client;
    }

    public void close() {
        CloseableUtils.closeQuietly(client);
    }

    public ConnectionStateListener getListener() {
        if (null != client) {
            ConnectionStateListener connectListener = new ConnectionStateListener() {
                @Override
                public void stateChanged(CuratorFramework curatorFramework, ConnectionState connectionState) {
                    if (connectionState == ConnectionState.LOST) {
                        while (true) {
                            try {
                                //手动重连
                                boolean flag = curatorFramework.getZookeeperClient().blockUntilConnectedOrTimedOut();
                                if (flag) {
                                    LOGGER.info("{}:[负载均衡修复]重连zk成功",ConnectionState.LOST);
                                    try {
                                        LOGGER.info("断开重连zookeeper");
                                        //根节点
//                                        Stat stat = client.checkExists().forPath(CONFIGURATION_ROOT_PATH);
//                                        if (null == stat) {
//                                            client.create().withMode(CreateMode.PERSISTENT).forPath(CONFIGURATION_ROOT_PATH);
//                                        }


                                        //service层部署到zookeeper
                                        if (RpcServiceZkFactory.getServiceZkDataList() != null && !RpcServiceZkFactory.getServiceZkDataList().isEmpty()) {
                                            LOGGER.info("service重连 :::" + JSON.toJSONString(RpcServiceZkFactory.getServiceZkDataList()));
                                            //注入节点信息
                                            RpcServiceZkFactory.getCommonServiceServer().registerServer(RpcServiceZkFactory.getServiceZkDataList());
                                        }
                                    } catch (InterruptedException e) {
                                        e.printStackTrace();
                                    } catch (Exception e) {
                                        e.printStackTrace();
                                    }
                                    break;
                                }
                            } catch (InterruptedException e) {
                                e.printStackTrace();
                            } catch (Exception e) {
                                e.printStackTrace();
                            }
                        }
                    } else if (connectionState == ConnectionState.RECONNECTED) {
                        //重新连接成功
                        //重新加载节点到zk
                        LOGGER.info("{}:[负载均衡修复]重连zk成功",ConnectionState.RECONNECTED);
                        // TODO: 2020/11/7  这里出现掉线情况需要重新加载
//                        try {
//                            LOGGER.info("断开重连zookeeper");
//                            //service层部署到zookeeper
//                            if (RpcServiceZkFactory.getServiceZkDataList() != null && !RpcServiceZkFactory.getServiceZkDataList().isEmpty()) {
//                                LOGGER.info("service重连 :::" + JSON.toJSONString(RpcServiceZkFactory.getServiceZkDataList()));
//                                //先删除之前的node，防止缓存，客户端感知不到
////                                        RpcServiceZkFactory.getCommonServiceServer().delServer(RpcServiceZkFactory.getServiceZkDataList());
//                                //注入节点信息
//                                RpcServiceZkFactory.getCommonServiceServer().registerServer(RpcServiceZkFactory.getServiceZkDataList());
//                            }
//                        } catch (InterruptedException e) {
//                            e.printStackTrace();
//                        } catch (Exception e) {
//                            e.printStackTrace();
//                        }
                    } else if (connectionState == ConnectionState.SUSPENDED) {
                        //自动重连,自动新建 schedular的临时节点
                        LOGGER.info("{}:自动重连,自动新建 schedular的临时节点",ConnectionState.SUSPENDED);

                    }
                }

            };
            return connectListener;
        }
        return null;
    }

}
